{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deterministic Terms in Time Series Models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "plt.rc(\"figure\", figsize=(16, 9))\n",
    "plt.rc(\"font\", size=16)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Basic Use\n",
    "\n",
    "Basic configurations can be directly constructed through `DeterministicProcess`. These can include a constant, a time trend of any order, and either a seasonal or a Fourier component.\n",
    "\n",
    "The process requires an index, which is the index of the full-sample (or in-sample).\n",
    "\n",
    "First, we initialize a deterministic process with a constant, a linear time trend, and a 5-period seasonal term. The `in_sample` method returns the full set of values that match the index."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from statsmodels.tsa.deterministic import DeterministicProcess\n",
    "\n",
    "index = pd.RangeIndex(0, 100)\n",
    "det_proc = DeterministicProcess(index, constant=True, order=1, seasonal=True, period=5)\n",
    "det_proc.in_sample()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `out_of_sample` returns the next `steps` values after the end of the in-sample."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.out_of_sample(15)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`range(start, stop)` can also be used to produce the deterministic terms over any range including in- and out-of-sample.\n",
    "\n",
    "### Notes\n",
    "\n",
    "* When the index is a pandas `DatetimeIndex` or a `PeriodIndex`, then `start` and `stop` can be date-like (strings, e.g., \"2020-06-01\", or Timestamp) or integers.\n",
    "* `stop` is always included in the range. While this is not very Pythonic, it is needed since both statsmodels and Pandas include `stop` when working with date-like slices."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.range(190, 210)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using a Date-like Index\n",
    "\n",
    "Next, we show the same steps using a `PeriodIndex`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "index = pd.period_range(\"2020-03-01\", freq=\"M\", periods=60)\n",
    "det_proc = DeterministicProcess(index, constant=True, fourier=2)\n",
    "det_proc.in_sample().head(12)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.out_of_sample(12)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`range` accepts date-like arguments, which are usually given as strings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.range(\"2025-01\", \"2026-01\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is equivalent to using the integer values 58 and 70."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.range(58, 70)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Advanced Construction\n",
    "\n",
    "Deterministic processes with features not supported directly through the constructor can be created using `additional_terms` which accepts a list of `DetermisticTerm`. Here we create a deterministic process with two seasonal components: day-of-week with a 5 day period and an annual captured through a Fourier component with a period of 365.25 days."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from statsmodels.tsa.deterministic import Fourier, Seasonality, TimeTrend\n",
    "\n",
    "index = pd.period_range(\"2020-03-01\", freq=\"D\", periods=2 * 365)\n",
    "tt = TimeTrend(constant=True)\n",
    "four = Fourier(period=365.25, order=2)\n",
    "seas = Seasonality(period=7)\n",
    "det_proc = DeterministicProcess(index, additional_terms=[tt, seas, four])\n",
    "det_proc.in_sample().head(28)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Custom Deterministic Terms\n",
    "\n",
    "The `DetermisticTerm` Abstract Base Class is designed to be subclassed to help users write custom deterministic terms.  We next show two examples. The first is a broken time trend that allows a break after a fixed number of periods. The second is a \"trick\" deterministic term that allows exogenous data, which is not really a deterministic process, to be treated as if was deterministic.  This lets use simplify gathering the terms needed for forecasting.\n",
    "\n",
    "These are intended to demonstrate the construction of custom terms. They can definitely be improved in terms of input validation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from statsmodels.tsa.deterministic import DeterministicTerm\n",
    "\n",
    "\n",
    "class BrokenTimeTrend(DeterministicTerm):\n",
    "    def __init__(self, break_period: int):\n",
    "        self._break_period = break_period\n",
    "\n",
    "    def __str__(self):\n",
    "        return \"Broken Time Trend\"\n",
    "\n",
    "    def _eq_attr(self):\n",
    "        return (self._break_period,)\n",
    "\n",
    "    def in_sample(self, index: pd.Index):\n",
    "        nobs = index.shape[0]\n",
    "        terms = np.zeros((nobs, 2))\n",
    "        terms[self._break_period :, 0] = 1\n",
    "        terms[self._break_period :, 1] = np.arange(self._break_period + 1, nobs + 1)\n",
    "        return pd.DataFrame(terms, columns=[\"const_break\", \"trend_break\"], index=index)\n",
    "\n",
    "    def out_of_sample(\n",
    "        self, steps: int, index: pd.Index, forecast_index: pd.Index = None\n",
    "    ):\n",
    "        # Always call extend index first\n",
    "        fcast_index = self._extend_index(index, steps, forecast_index)\n",
    "        nobs = index.shape[0]\n",
    "        terms = np.zeros((steps, 2))\n",
    "        # Assume break period is in-sample\n",
    "        terms[:, 0] = 1\n",
    "        terms[:, 1] = np.arange(nobs + 1, nobs + steps + 1)\n",
    "        return pd.DataFrame(\n",
    "            terms, columns=[\"const_break\", \"trend_break\"], index=fcast_index\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "btt = BrokenTimeTrend(60)\n",
    "tt = TimeTrend(constant=True, order=1)\n",
    "index = pd.RangeIndex(100)\n",
    "det_proc = DeterministicProcess(index, additional_terms=[tt, btt])\n",
    "det_proc.range(55, 65)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we write a simple \"wrapper\" for some actual exogenous data that simplifies constructing out-of-sample exogenous arrays for forecasting."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ExogenousProcess(DeterministicTerm):\n",
    "    def __init__(self, data):\n",
    "        self._data = data\n",
    "\n",
    "    def __str__(self):\n",
    "        return \"Custom Exog Process\"\n",
    "\n",
    "    def _eq_attr(self):\n",
    "        return (id(self._data),)\n",
    "\n",
    "    def in_sample(self, index: pd.Index):\n",
    "        return self._data.loc[index]\n",
    "\n",
    "    def out_of_sample(\n",
    "        self, steps: int, index: pd.Index, forecast_index: pd.Index = None\n",
    "    ):\n",
    "        forecast_index = self._extend_index(index, steps, forecast_index)\n",
    "        return self._data.loc[forecast_index]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "gen = np.random.default_rng(98765432101234567890)\n",
    "exog = pd.DataFrame(gen.integers(100, size=(300, 2)), columns=[\"exog1\", \"exog2\"])\n",
    "exog.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ep = ExogenousProcess(exog)\n",
    "tt = TimeTrend(constant=True, order=1)\n",
    "# The in-sample index\n",
    "idx = exog.index[:200]\n",
    "det_proc = DeterministicProcess(idx, additional_terms=[tt, ep])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.in_sample().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "det_proc.out_of_sample(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model Support\n",
    "\n",
    "The only model that directly supports `DeterministicProcess` is `AutoReg`. A custom term can be set using the `deterministic` keyword argument. \n",
    "\n",
    "**Note**: Using a custom term requires that `trend=\"n\"` and `seasonal=False` so that all deterministic components must come from the custom deterministic term."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simulate Some Data\n",
    "\n",
    "Here we simulate some data that has an weekly seasonality captured by a Fourier series."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "gen = np.random.default_rng(98765432101234567890)\n",
    "idx = pd.RangeIndex(200)\n",
    "det_proc = DeterministicProcess(idx, constant=True, period=52, fourier=2)\n",
    "det_terms = det_proc.in_sample().to_numpy()\n",
    "params = np.array([1.0, 3, -1, 4, -2])\n",
    "exog = det_terms @ params\n",
    "y = np.empty(200)\n",
    "y[0] = det_terms[0] @ params + gen.standard_normal()\n",
    "for i in range(1, 200):\n",
    "    y[i] = 0.9 * y[i - 1] + det_terms[i] @ params + gen.standard_normal()\n",
    "y = pd.Series(y, index=idx)\n",
    "ax = y.plot()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The model is then fit using the `deterministic` keyword argument. `seasonal` defaults to False but `trend` defaults to `\"c\"` so this needs to be changed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from statsmodels.tsa.api import AutoReg\n",
    "\n",
    "mod = AutoReg(y, 1, trend=\"n\", deterministic=det_proc)\n",
    "res = mod.fit()\n",
    "print(res.summary())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use the `plot_predict` to show the predicted values and their prediction interval. The out-of-sample deterministic values are automatically produced by the deterministic process passed to `AutoReg`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fig = res.plot_predict(200, 200 + 2 * 52, True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "auto_reg_forecast = res.predict(200, 211)\n",
    "auto_reg_forecast"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using with other models\n",
    "\n",
    "Other models do not support `DeterministicProcess` directly.  We can instead manually pass any deterministic terms as `exog` to model that support exogenous values.\n",
    "\n",
    "Note that `SARIMAX` with exogenous variables is OLS with SARIMA errors so that the model is \n",
    "\n",
    "$$\n",
    "\\begin{align*}\n",
    "\\nu_t & = y_t - x_t \\beta  \\\\\n",
    "(1-\\phi(L))\\nu_t & = (1+\\theta(L))\\epsilon_t.\n",
    "\\end{align*}\n",
    "$$\n",
    "\n",
    "The parameters on deterministic terms are not directly comparable to `AutoReg` which evolves according to the equation\n",
    "\n",
    "$$\n",
    "(1-\\phi(L)) y_t = x_t \\beta + \\epsilon_t.\n",
    "$$\n",
    "\n",
    "When $x_t$ contains only deterministic terms, these two representation are equivalent (assuming $\\theta(L)=0$ so that there is no MA).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from statsmodels.tsa.api import SARIMAX\n",
    "\n",
    "det_proc = DeterministicProcess(idx, period=52, fourier=2)\n",
    "det_terms = det_proc.in_sample()\n",
    "\n",
    "mod = SARIMAX(y, order=(1, 0, 0), trend=\"c\", exog=det_terms)\n",
    "res = mod.fit(disp=False)\n",
    "print(res.summary())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The forecasts are similar but differ since the parameters of the `SARIMAX` are estimated using MLE while `AutoReg` uses OLS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sarimax_forecast = res.forecast(12, exog=det_proc.out_of_sample(12))\n",
    "df = pd.concat([auto_reg_forecast, sarimax_forecast], axis=1)\n",
    "df.columns = columns = [\"AutoReg\", \"SARIMAX\"]\n",
    "df"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
